package SparkApps;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

import java.util.Arrays;
import java.util.List;

public class SparkDemo {
    public static void main(String args[]) {
        SparkConf sparkConf = new SparkConf().setAppName("JavaSparkDemo").setMaster("yarn-client");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        List<Integer> data1 = Arrays.asList(1, 2, 3, 4, 5, 6, 6);
        List<Integer> data2 = Arrays.asList(9, 8, 7, 6, 5, 4, 3);
        JavaRDD<Integer> rdd1 = sc.parallelize(data1);
        JavaRDD<Integer> rdd2 = sc.parallelize(data2);

        /*------------这里是对单个RDD的转化操作-------------*/
        JavaRDD<Integer> afterMap = rdd1.map(x -> x+1);
        System.out.println("afterMap---" + afterMap.collect().toString());//afterMap---[2, 3, 4, 5, 6, 7, 7]

        JavaRDD<Integer> afterFlatMap = rdd1.map(x -> x-1);
        System.out.println("afterFlatMap---" + afterFlatMap.collect().toString());//afterFlatMap---[0, 1, 2, 3, 4, 5, 5]

        JavaRDD<Integer> afterFilter = rdd1.filter(x -> x>2);
        System.out.println("afterFilter---" + afterFilter.collect().toString());//afterFilter---[3, 4, 5, 6, 6]

        JavaRDD<Integer> afterDisTinct = rdd1.distinct();
        System.out.println("afterDisTinct---" + afterDisTinct.collect().toString());//afterDisTinct---[4, 6, 2, 1, 3, 5]

        /*------------这里是对两个RDD的转化操作-------------*/
        JavaRDD<Integer> afterUnion = rdd1.union(rdd2);
        System.out.println("afterUnion---" + afterUnion.collect().toString());//afterUnion---[1, 2, 3, 4, 5, 6, 6, 9, 8, 7, 6, 5, 4, 3]

        JavaRDD<Integer> afterIntersection = rdd1.intersection(rdd2);
        System.out.println("afterIntersection---" + afterIntersection.collect().toString());//afterIntersection---[4, 6, 3, 5]

        JavaRDD<Integer> afterSubtarct = rdd1.subtract(rdd2);
        System.out.println("afterSubtarct---" + afterSubtarct.collect().toString());//afterSubtarct---[2, 1]

        /*------------这里是对RDD的行为(action)操作----------*/
        System.out.println(rdd1.collect());//[1, 2, 3, 4, 5, 6, 6]
        System.out.println(rdd1.count());//7
        System.out.println(rdd1.countByValue());//{5=1, 1=1, 6=2, 2=1, 3=1, 4=1}
        System.out.println(rdd1.take(1));//[1]
        System.out.println(rdd1.top(2));//[6, 6]
        System.out.println(rdd1.takeOrdered(4));//[1, 2, 3, 4]
        System.out.println(rdd1.takeSample(false, 1));//[6]
        System.out.println(rdd1.reduce((x, y) -> x+y));//27
        System.out.println(rdd1.fold(1, (x, y) -> x+y));//30
        rdd1.foreach(x -> System.out.println(x));//这里没有输出可能这个表达有点问题
        sc.stop();
    }
}
